package com.lianda.operator;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Map算子
 */
public class MapMain {
    public static void main(String[] args) throws Exception {

        //监听端口
        String hostname = "127.0.0.1";
        Integer port = Integer.parseInt("9000");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream = env.socketTextStream(hostname, port);

        //完成map的转换
        SingleOutputStreamOperator<String> output = stream.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                String ret = null;
                if ("Jack".equalsIgnoreCase(s)) {
                    ret = "Hello, " + s;
                } else {
                    ret = s;
                }
                return ret;
            }
        });

        output.print();
        env.execute("demo");
    }
}
